package org.senRPC.client.provider.controler;

import static org.senRPC.common.serialization.SerializerHolder.serializerImpl;
import static org.senRPC.common.utils.Reflects.fastInvoke;
import static org.senRPC.common.utils.Reflects.findMatchingParameterTypes;
import static org.senRPC.common.utils.Status.APP_FLOW_CONTROL;
import static org.senRPC.common.utils.Status.BAD_REQUEST;
import static org.senRPC.common.utils.Status.SERVICE_NOT_FOUND;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;

import java.util.List;

import org.senRPC.client.metrics.ServiceMeterManager;
import org.senRPC.client.provider.container.DefaultServiceProviderContainer.CurrentServiceState;

import org.senRPC.client.provider.model.ServiceWrapper;
import org.senRPC.common.protocal.senRPCProtocol;
import org.senRPC.common.transport.body.RequestCustomBody;
import org.senRPC.common.transport.body.ResponseCustomBody;
import org.senRPC.common.transport.body.ResponseCustomBody.ResultWrapper;
import org.senRPC.common.utils.Pair;
import org.senRPC.common.utils.Status;
import org.senRPC.common.utils.SystemClock;
import org.senRPC.remoting.model.RemotingTransporter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author senrian
 * @description 处理consumer rpc请求的核心控制器，并统计处理的次数
 * @time 2020年2月30日
 * @modifytime
 */
public class ProviderRPCController {

	private static final Logger logger = LoggerFactory.getLogger(ProviderRPCController.class);

	private DefaultProvider defaultProvider;

	public ProviderRPCController(DefaultProvider defaultProvider) {
		this.defaultProvider = defaultProvider;
	}
	//接收到一个RemotingTransporter
	public void handlerRPCRequest(RemotingTransporter request, Channel channel) {


		String serviceName = null;
		RequestCustomBody body = null;
		int requestSize = 0;

		try {
			byte[] bytes = request.bytes();
			requestSize = bytes.length;
			request.bytes(null);
			//反序列化得到真正的body
			body = serializerImpl().readObject(bytes, RequestCustomBody.class);

			request.setCustomHeader(body);
			serviceName = body.getServiceName();

			ServiceMeterManager.incrementRequestSize(serviceName, requestSize);
			ServiceMeterManager.incrementCallTimes(serviceName);

		} catch (Exception e) {
			rejected(BAD_REQUEST, channel, request,serviceName);
			return;
		}
		//从Container中获取服务pair， state是服务控制相关，wrapper是业务相关
		final Pair<CurrentServiceState, ServiceWrapper> pair = defaultProvider.getRegistryController().getProviderContainer().lookupService(serviceName);
		if (pair == null || pair.getValue() == null) {
            rejected(SERVICE_NOT_FOUND, channel, request,serviceName);
            return;
        }

		// app flow control
		if(pair.getValue().isFlowController()){

	        ServiceFlowControllerManager serviceFlowControllerManager = defaultProvider.getRegistryController().getServiceFlowControllerManager();
	        if (!serviceFlowControllerManager.isAllow(serviceName)) {
	            rejected(APP_FLOW_CONTROL,channel, request,serviceName);
	            return;
	        }
		}


        process(pair,request,channel,serviceName,body.getTimestamp());
	}



	/**
	 * RPC的核心处理
	 * @param pair
	 * @param request
	 * @param channel
	 * @param serviceName
	 * @param beginTime
	 */
	//现在已经拿到了方法名，入参，接下来就是调用一些反射的API,就可以完成了方法的调用了：
	private void process(Pair<CurrentServiceState, ServiceWrapper> pair, final RemotingTransporter request, Channel channel,final String serviceName,final long beginTime) {

		Object invokeResult = null;

		CurrentServiceState currentServiceState = pair.getKey();
		ServiceWrapper serviceWrapper = pair.getValue();
		//服务端目标方法。
		Object targetCallObj = serviceWrapper.getServiceProvider();
		Object[] args = ((RequestCustomBody)request.getCustomHeader()).getArgs();

		//判断服务是否已经被设定为自动降级，如果被设置为自动降级且有它自己的mock类的话，则将targetCallObj切换到mock方法上来
		if(currentServiceState.getHasDegrade().get() && serviceWrapper.getMockDegradeServiceProvider() != null){
			targetCallObj = serviceWrapper.getMockDegradeServiceProvider();
		}
		//方法名
		String methodName = serviceWrapper.getMethodName();
		//参数类型
		List<Class<?>[]> parameterTypesList = serviceWrapper.getParamters();


		Class<?>[] parameterTypes = findMatchingParameterTypes(parameterTypesList, args);
		//方法体,
		invokeResult = fastInvoke(targetCallObj, methodName, parameterTypes, args);

		ResultWrapper result = new ResultWrapper();
		result.setResult(invokeResult);
		ResponseCustomBody body = new ResponseCustomBody(Status.OK.value(), result);

		final RemotingTransporter response = RemotingTransporter.createResponseTransporter(senRPCProtocol.RPC_RESPONSE, body, request.getOpaque());

		channel.writeAndFlush(response).addListener(new ChannelFutureListener() {

			public void operationComplete(ChannelFuture future) throws Exception {

				long elapsed = SystemClock.millisClock().now() - beginTime;

				if (future.isSuccess()) {
				    //累加某个服务的调用时间
					ServiceMeterManager.incrementTotalTime(serviceName, elapsed);
				} else {
					logger.info("request {} get failed response {}", request, response);
				}
			}
		});

	}

	private void rejected(Status status, Channel channel, final RemotingTransporter request,String serviceName) {

		if(null != serviceName){
			ServiceMeterManager.incrementFailTimes(serviceName);
		}
		ResultWrapper result = new ResultWrapper();
		switch (status) {
		case BAD_REQUEST:
			result.setError("bad request");
		case SERVICE_NOT_FOUND:
			result.setError(((RequestCustomBody) request.getCustomHeader()).getServiceName() +" no service found");
			break;
		case APP_FLOW_CONTROL:
		case PROVIDER_FLOW_CONTROL:
			result.setError("over unit time call limit");
			break;
		default:
			logger.warn("Unexpected status.", status.description());
			return;
		}
		logger.warn("Service rejected: {}.", result.getError());

		ResponseCustomBody responseCustomBody = new ResponseCustomBody(status.value(), result);
		final RemotingTransporter response = RemotingTransporter.createResponseTransporter(senRPCProtocol.RPC_RESPONSE, responseCustomBody,
				request.getOpaque());

		channel.writeAndFlush(response).addListener(new ChannelFutureListener() {

			public void operationComplete(ChannelFuture future) throws Exception {
				if (future.isSuccess()) {
					logger.info("request error {} get success response {}", request, response);
				} else {
					logger.info("request error {} get failed response {}", request, response);
				}
			}
		});
	}

}
